package com.junmo.www.RxJava;

import io.reactivex.Observable;
import io.reactivex.ObservableEmitter;
import io.reactivex.ObservableOnSubscribe;
import io.reactivex.annotations.NonNull;
import io.reactivex.functions.Consumer;
import io.reactivex.functions.Function;
import io.reactivex.schedulers.Schedulers;
import org.testng.annotations.Test;


public class SchedulerTest {
    @Test
    public void test() {

    }
    @Test
    public void test03() {
        Observable.just("HELLO WORLD")
                .subscribeOn(Schedulers.single())
                .map(new Function<String, String>() {
                    @Override
                    public String apply(@NonNull String s) throws Exception {
                        s = s . toLowerCase ();
                        System.out.format("map1--%s\n", s );
                        return s ;
                    }
                })
                .observeOn(Schedulers.io())
                .map(new Function<String, String>() {
                    @Override
                    public String apply(@NonNull String s) throws Exception {
                        s = s +" tony. ";
                        System.out.format("map2--%s\n", s) ;
                        return s;
                    }
                })
                .subscribeOn(Schedulers.computation())
                .map(new Function<String, String>() {
                    @Override
                    public String apply(@NonNull String s) throws Exception {
                        s = s +"it is a test. ";
                        System.out.format("map3--%s\n", s);
                        return s;
                    }
                })
                .observeOn(Schedulers.newThread())
                .subscribe(new Consumer<String>() {
                    @Override
                    public void accept(String s) throws Exception {
                        System.out.format("subscribe--%s\n", s) ;
                    }
                });
    }

    @Test
    public void test02() {
        Observable.create(new ObservableOnSubscribe<String>() {
            @Override
            public void subscribe(@NonNull ObservableEmitter<String> e) throws Exception {
                e.onNext ("aaa");
                e.onNext ("bbb");
            }
        }).observeOn(Schedulers.newThread())
        .map(new Function<String, String>() {
            @Override
            public String apply(@NonNull String s) throws Exception {
                return s.toUpperCase();
            }
        })
        .subscribeOn(Schedulers.single())
        .observeOn(Schedulers.io())
        .subscribe(new Consumer<String>() {
            @Override
            public void accept(String s) throws Exception {
                System.out.println (s);
                System.out.println(Thread.currentThread().getId());
            }
        });
    }


    @Test
    public void test01() {
        Observable.create(new ObservableOnSubscribe<String>() {
            @Override
            public void subscribe(@NonNull ObservableEmitter<String> e) throws Exception {
                e.onNext ("hello");
                e.onNext ("world");
                e.onNext ("hello22");
                e.onNext ("world22");
            }
        }).observeOn(Schedulers.newThread())
        .subscribe(new Consumer<String>() {
            @Override
            public void accept(String s) throws Exception {
                System.out.println (s);
                System.out.println(Thread.currentThread().getId());
            }
        });
    }
}
